package dao

import (
	"context"
	"strings"

	"github.com/nsqio/go-nsq"

	"open/pkg/log"
)

type nsqLogger struct{}

// Output impl
func (n *nsqLogger) Output(_ int, s string) (err error) {
	if strings.Contains(s, "404 Not Found") {
		return nil
	}
	log.FromContext(context.Background()).Info(s)
	return
}

// RunWorker 增加一个worker
func (d *Dao) RunWorker(ctx context.Context, topic string, handler nsq.Handler) {
	logger := log.FromContext(ctx)
	consumer, err := nsq.NewConsumer(topic, d.conf.Nsq.Channel, nsq.NewConfig())
	if err != nil {
		logger.Fatalf("could not create instance of Consumer with topic: %s , err: %v", topic, err)
		return
	}
	consumer.SetLogger(&nsqLogger{}, nsq.LogLevelError)
	consumer.AddConcurrentHandlers(handler, 5)
	err = consumer.ConnectToNSQLookupd(d.conf.Nsq.LookupAddress)
	if err != nil {
		logger.Fatal("could not connect to nsq lookup addr, err: %v", err)
		return
	}
	// saving for close
	d.consumers = append(d.consumers, consumer)
}
